package com.wolf.etl.core.task;

import com.wolf.etl.core.ETLManager;
import com.wolf.etl.model.EtlTaskModel;
import com.wolf.etl.service.IEtlTaskService;
import org.apache.commons.lang3.StringUtils;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.quartz.impl.matchers.GroupMatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.List;
import java.util.Set;

/**
 * @author sdyang
 * @date 2019/11/4 8:19
 **/
@Service("ScheduleJobService")
public class ScheduleJobService {

    private static Logger logger = LoggerFactory.getLogger(ScheduleJobService.class);

    // 获取工厂类
    private StdSchedulerFactory sf = new StdSchedulerFactory();

    @Autowired
    private IEtlTaskService etlTaskService;
    @Autowired
    private ETLManager etlManager;

    // 项目重启后，初始化原本已经运行的定时任务
    public void init() {
        List<EtlTaskModel> vos = etlManager.findTaskByStatus(TaskStatus.ENABLE);

        vos.forEach(vo -> {
            startScheduleByInit(vo);
        });
    }

    /**
     * 初始化时开启定时任务
     */
    private void startScheduleByInit(EtlTaskModel task) {
        try {
            Scheduler scheduler = sf.getScheduler();
            startJob(scheduler, task);
            scheduler.start();
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }
    }

    /**
     * 开启定时任务
     *
     * @param model
     */
    public void start(EtlTaskModel model) {
        if (StringUtils.isEmpty(model.getGroup_name()) || StringUtils.isEmpty(model.getTask_name()) || StringUtils.isEmpty(model.getCron())) {
            throw new RuntimeException("参数不能为空");
        }
        List<EtlTaskModel> poList = etlTaskService.find(model.getGroup_name(), model.getTask_name(), TaskStatus.ENABLE.name());
        if (!ObjectUtils.isEmpty(poList)) {
            throw new RuntimeException("group和job名称已存在");
        }
        try {
            Scheduler scheduler = sf.getScheduler();
            startJob(scheduler, model);
            scheduler.start();
//            EtlTaskModel task = new EtlTaskModel();
//            task.setGroup_name(model.getGroup_name());
//            task.setTask_name(model.getTask_name());
//            task.setCron(model.getCron());
//            task.setStatus(TaskStatus.ENABLE);
//            etlTaskService.insertModel(task);
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }

    }

    /**
     * 更新定时任务
     *
     * @param model
     */
    public void updateCorn(EtlTaskModel model) {
        if (ObjectUtils.isEmpty(model.getId()) || ObjectUtils.isEmpty(model.getCron())) {
            throw new RuntimeException("定时任务不存在");
        }
        try {
            EtlTaskModel po = etlTaskService.find(model.getId(), TaskStatus.ENABLE.name());
            // 获取调度对象
            Scheduler scheduler = sf.getScheduler();
            // 获取触发器
            TriggerKey triggerKey = new TriggerKey(po.getTask_name(), po.getGroup_name());
            CronTrigger cronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);
            String oldTime = cronTrigger.getCronExpression();
            if (!oldTime.equalsIgnoreCase(model.getCron())) {
                CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(model.getCron());
                CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(po.getTask_name(), po.getGroup_name())
                        .withSchedule(cronScheduleBuilder).build();
                // 更新定时任务
                scheduler.rescheduleJob(triggerKey, trigger);
                po.setCron(model.getCron());
                // 更新数据库
                etlTaskService.updateModel(po);
            }
        } catch (Exception e) {
            logger.info("exception:{}", e);
        }

    }

    /**
     * 任务 - 暂停
     */
    public void pause(EtlTaskModel model) {
        if (ObjectUtils.isEmpty(model.getId())) {
            throw new RuntimeException("定时任务不存在");
        }
        EtlTaskModel po = etlTaskService.find(model.getId(), TaskStatus.ENABLE.name());
        if (ObjectUtils.isEmpty(po)) {
            throw new RuntimeException("定时任务不存在");
        }
        try {
            Scheduler scheduler = sf.getScheduler();
            JobKey jobKey = new JobKey(po.getTask_name(), po.getGroup_name());
            JobDetail jobDetail = scheduler.getJobDetail(jobKey);
            if (jobDetail == null)
                return;
            scheduler.pauseJob(jobKey);
            po.setStatus(TaskStatus.PAUSED.name());
            etlTaskService.updateModel(po);
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }
    }

    /**
     * 任务 - 恢复
     */

    public void resume(EtlTaskModel model) {
        if (ObjectUtils.isEmpty(model.getId())) {
            throw new RuntimeException("定时任务不存在");
        }
        EtlTaskModel po = etlTaskService.find(model.getId(), TaskStatus.PAUSED.toString());
        if (ObjectUtils.isEmpty(po)) {
            throw new RuntimeException("定时任务不存在");
        }
        try {
            Scheduler scheduler = sf.getScheduler();
            JobKey jobKey = new JobKey(po.getTask_name(), po.getGroup_name());
            JobDetail jobDetail = scheduler.getJobDetail(jobKey);
            if (jobDetail == null)
                return;
            scheduler.resumeJob(jobKey);
            po.setStatus(TaskStatus.ENABLE.name());
            etlTaskService.updateModel(po);
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }
    }

    /**
     * 任务 - 删除一个定时任务
     */
    public void disable(EtlTaskModel model) {
        if (ObjectUtils.isEmpty(model.getId())) {
            throw new RuntimeException("定时任务不存在");
        }
        EtlTaskModel po = etlTaskService.find(model.getId(), TaskStatus.ENABLE.toString());
        if (ObjectUtils.isEmpty(po)) {
            throw new RuntimeException("定时任务不存在");
        }
        try {
            Scheduler scheduler = sf.getScheduler();
            JobKey jobKey = new JobKey(po.getTask_name(), po.getGroup_name());
            JobDetail jobDetail = scheduler.getJobDetail(jobKey);
            if (jobDetail == null)
                return;
            scheduler.deleteJob(jobKey);
            po.setStatus(TaskStatus.DISABLE.toString());
            etlTaskService.updateModel(po);
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }
    }

    /**
     * 删除所有定时任务
     */
    public void disableAll() {
        try {
            Scheduler scheduler = sf.getScheduler();
            // 获取有所的组
            List<String> jobGroupNameList = scheduler.getJobGroupNames();
            for (String jobGroupName : jobGroupNameList) {
                GroupMatcher<JobKey> jobKeyGroupMatcher = GroupMatcher.jobGroupEquals(jobGroupName);
                Set<JobKey> jobKeySet = scheduler.getJobKeys(jobKeyGroupMatcher);
                for (JobKey jobKey : jobKeySet) {
                    String jobName = jobKey.getName();
                    JobDetail jobDetail = scheduler.getJobDetail(jobKey);
                    if (jobDetail == null)
                        return;
                    scheduler.deleteJob(jobKey);
                    // 更新数据库
                    List<EtlTaskModel> poList = etlTaskService.find(jobGroupName, jobName, TaskStatus.ENABLE.toString());
                    poList.forEach(po -> {
                        po.setStatus(TaskStatus.DISABLE.toString());
                        etlTaskService.updateModel(po);
                    });
                    logger.info("group:{}, job:{}", jobGroupName, jobName);
                }
            }
        } catch (Exception e) {
            logger.error("exception:{}", e);
        }
    }

    // 开启任务
    private void startJob(Scheduler scheduler, EtlTaskModel task) throws SchedulerException {

        String group = task.getGroup_name();
        String name = task.getTask_name();
        String cron = task.getCron();
        Long task_id = task.getId();
        // 通过JobBuilder构建JobDetail实例，JobDetail规定只能是实现Job接口的实例
        // 在map中可传入自定义参数，在job中使用
        JobDataMap map = new JobDataMap();
        map.put("group", group);
        map.put("name", name);
        map.put("task_id",task_id);
        map.put("manager",etlManager);
        // JobDetail 是具体Job实例
        JobDetail jobDetail = JobBuilder.newJob(ScheduleQuartzJob.class).withIdentity(name, group)
                .usingJobData(map)
                .build();
        // 基于表达式构建触发器
        CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(cron);
        // CronTrigger表达式触发器 继承于Trigger
        // TriggerBuilder 用于构建触发器实例
        CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(name, group)
                .withSchedule(cronScheduleBuilder).build();
        scheduler.scheduleJob(jobDetail, cronTrigger);
    }
}
